package com.mq.mqserver;
/*
* 通过这个类，来表示 虚拟主机
* 每个虚拟主机下面都管理着自己的交换机，队列，绑定，消息数据
* 同时提供 api 供上层调用
* */

import com.mq.common.exception.MqException;
import com.mq.mqserver.core.*;
import com.mq.mqserver.datacenter.DiskDataCenter;
import com.mq.mqserver.datacenter.MemoryDataCenter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import com.mq.common.Consumer;
public class VirtualHost {
    private String virtualHostName;
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    private Router router = new Router();
    private ConsumerManager consumerManager = new ConsumerManager(this);
    //操作交换机锁对象
    private final Object exchangeLocker = new Object();
    //操作队列锁对象
    private final Object queueLocker = new Object();
    //操作绑定锁对象
    private final Object bindingLocker = new Object();
    public String getVirtualHostName() {
        return virtualHostName;
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }

    public VirtualHost(String name){
        this.virtualHostName = name;

        //对于memoryDataCenter初始化，只需要new出来
        //对于diskDataCenter初始化，需要init()
        diskDataCenter.init();
        //针对硬盘数据，回复到内存
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException |MqException|ClassNotFoundException e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 恢复内存数据失败");
        }
    }
    //创建换机
    //如果交换机不存在，就创建，如果存在就返回
    //返回值是 bolean ,创建成功返回true，失败返回false
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable,
                                   boolean autoDelete, Map<String,Object> arguments){
        //把交换机的名字加上虚拟机主机作为前缀
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                //1.判定该交换机是否已经存在，直接在内存中查询
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if (existsExchange != null) {
                    //该交换机已存在
                    System.out.println("[VirtualHost] 交换机已经存在！ exchangeName=" + exchangeName);
                    return true;
                }
                //2.真正创建交换机，先构建Exchange对象
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);
                //3.把交换机对象写入硬盘
                if (durable) {
                    diskDataCenter.insertExchange(exchange);
                }
                //4.把交换机对象写入内存
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost] 交换机创建完成！ exchangeName=" + exchangeName);
                //先写磁盘再写内存，因为磁盘容易写失败，磁盘写失败了，内存就不写了
                //要是先写内存，内存成功了。硬盘失败了，还要把内存的数据再删掉
                return true;
            }
        }catch (Exception e){
            System.out.println("[VirtualHost] 交换机创建失败！ exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }

    }
    //删除交换机
    public boolean exchangeDelete(String exchangeName){
        exchangeName = virtualHostName + exchangeName;
        try{
            synchronized (exchangeLocker) {
                //1.找到对应的交换机
                Exchange existExchanage = memoryDataCenter.getExchange(exchangeName);
                if (existExchanage == null) {
                    throw new MqException("[VirtualHost] 交换机不存在无法删除！");
                }
                //2.删除硬盘上数据
                if (existExchanage.isDurable()) {
                    diskDataCenter.deleteExchange(exchangeName);
                }
                //删除内存数据
                memoryDataCenter.deleteExchange(exchangeName);
                System.out.println("[VirtualHost] 交换机删除成功！ exchangeName=" + exchangeName);
                return true;
            }
        }catch (Exception e){
            System.out.println("[VirtualHost] 交换机删除失败！ exchangeName=" + exchangeName);
            e.printStackTrace();
            return  false;
        }
    }

    //创建队列
    public boolean queueDeclare(String queueName,boolean durable,boolean exclusive,boolean autoDelete,
                                Map<String,Object> arguements){
        queueName = virtualHostName + queueName;
        try{
            synchronized (queueLocker) {
                MSGQueue existQueue = memoryDataCenter.getQueue(queueName);
                //1.先查一查队列是否存在
                if (existQueue != null) {
                    throw new Exception("[VirtualHost] 队列已经存在！ queueName =" + queueName);
                }
                //2.不存在创建新的队列
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguements);
                //3.队列写入磁盘
                if (durable) {
                    diskDataCenter.insertQueue(queue);
                }
                //4.队列写入内存
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 队列创建完成！ queueName=" + queueName);
                return true;
            }
        }catch (Exception e){
            System.out.println("[VirtualHost] 队列创建失败！ queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }
    //删除队列
    public boolean queueDelete(String queueName){
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                MSGQueue exietQueue = memoryDataCenter.getQueue(queueName);
                if (exietQueue == null) {
                    throw new MqException("[VirtualHost] 队列不存在！ queueName =" + queueName);
                }
                if (exietQueue.isDurable()) {
                    diskDataCenter.deleteQueue(queueName);
                }
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost] 队列删除完成！ queueName=" + queueName);
                return true;
            }
        }catch (Exception e){
            System.out.println("[VirtualHost] 队列删除失败！ queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }
    //创建绑定
    public boolean queueBind(String queueName,String exchangeName,String bindingKey){
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try{
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    //1.查询当前绑定是否已经存在
                    Binding existBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (existBinding != null) {
                        throw new MqException("[VirtualHost] 绑定已存在！ queueName =" + queueName + "exchangeName = " + exchangeName);
                    }
                    //2.验证BindingKey是否合法
                    if (!router.checkBindingKey(bindingKey)) {
                        throw new MqException("[VirtualHost] bindingKey非法 bindingKey = " + bindingKey);
                    }

                    //3.创建binding 对象
                    Binding binding = new Binding();
                    binding.setQueueName(queueName);
                    binding.setExchangeName(exchangeName);
                    binding.setBindingKey(bindingKey);

                    //4.获取对应的交换机和队列，要数不存在就无法创建绑定
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if (queue == null) {
                        throw new MqException("[VirtualHost] 队列不存在 queueName = " + queueName);
                    }
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if (exchange == null) {
                        throw new MqException("[VirtualHost] 交换机不存在 exchangeName = " + exchangeName);
                    }
                    //5.先写硬盘
                    if (queue.isDurable() && exchange.isDurable()) {
                        diskDataCenter.insertBinding(binding);
                    }
                    //6.写入内存
                    memoryDataCenter.insertBindingExchangeName(binding);
                    System.out.println("[VirtualHost] 绑定创建成功 exchangeName = " + exchangeName + "queueName = " + queueName);
                    return true;
                }
            }

        }catch (Exception e){
            System.out.println("[VirtualHost] queueBind 失败");
            e.printStackTrace();
            return false;
        }
    }
    //删除绑定
    public boolean queueUnbind (String queueName,String exchangeName){
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try{
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    //1.获取绑定
                    Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (binding == null) {
                        throw new MqException("[VirtualHost] 删除绑定失败，绑定不存在 queueName = " + queueName + "exchangeName = " + exchangeName);
                    }
                    /*//2.获取一下对应的队列和交换机，看是否存在
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if(queue == null){
                        throw new MqException("[VirtualHost] 删除绑定失败，对应队列不存在 queueName = "+ queueName );
                    }
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if(exchange == null){
                        throw new MqException("[VirtualHost] 删除绑定失败，对应交换机不存在 exchange = "+ exchange );
                    }
                    //3.删除硬盘队列
                    if(queue.isDurable() && exchange.isDurable()){
                        diskDataCenter.deleteBinding(binding);
                    }*/
                    //2.无论绑定是否持久化，都尝试从硬盘中删除，就算不存在，也没有副作用
                    diskDataCenter.deleteBinding(binding);
                    memoryDataCenter.deleteBinding(binding);
                    System.out.println("[VirtualHost] 删除绑定成功 ");
                    return true;
                }
            }
        }catch (Exception e){
            System.out.println("[VirtualHost] 删除绑定失败 ");
            e.printStackTrace();
            return false;
        }
    }
    //1、以上加得锁，针对A交换机的操作会影响到B交换机的操作，可以调整
    //但是影响不大，因为对于 Broker Server 来说，创建交换机，队列，绑定是低频操作
    //低频操作，所以遇到两个线程都误操作创建队列之类的情况概率就会低
    //因此大多数不会触发锁冲突，只有遇到真竞争才加锁

    //2、这里既然枷锁了，那memoryDataCenter是不是不用加锁
    //因为 memoryDataCenter 可能给其他类调用

    //发送消息到指定的交换机/绑定
    public boolean basicPublish(String exchangeName,String routingKey,BasicProperties basicProperties,byte[] body){
        try{
            //1.转换交换机名字
            exchangeName = virtualHostName + exchangeName;
            //2.检查routingKey是否合法
            if(!router.checkRoutingKey(routingKey)){
                throw new MqException("[VirtualHost] 发送信息失败，routingKey非法 routingKey = " + routingKey);
            }
            //3。查找交换机对象
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if(exchange == null){
                throw new MqException("[VirtualHost] 发送信息失败，交换机不存在 exchangeName = " + exchangeName);
            }
            //4.判定交换机类型
            if(exchange.getType() == ExchangeType.DIRECT){
                //按照直接交换机的方式进行转发消息
                //以 routingKey 作为队列名字。直接写入指定的队列
                //此时可以无视绑定关系，比如 exchange 和 queue 之间美誉绑定，
                //那么可以通过 routingKey中设定的队列名实现消息的发送，最常用，简单

                String queueName = virtualHostName + routingKey;
                //构建一个新的消息
                Message message = Message.createMessageWithId(routingKey,basicProperties,body);
                //查询队列是否存在
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if(queue == null){
                    throw new MqException("[VirtualHost] 发送信息失败，ExchangeType是DIRECT 队列不存在 queueName = " + queueName);
                }
                //队列存在写入消息
                sendMessage(queue,message);
                System.out.println("[VirtualHost] 发送信息成功");

            }else {
                //按照 fanout 和 topic 方式来转发
                //找到该交换机关联的所有绑定，并遍历这些绑定对象
                ConcurrentHashMap <String,Binding> bindingMap = memoryDataCenter.getBindings(exchangeName);
                for(Map.Entry<String,Binding> entry: bindingMap.entrySet()){
                    //获取绑定对象，判定对应的队列是否存在
                    Binding binding = entry.getValue();
                    MSGQueue queue =  memoryDataCenter.getQueue(binding.getQueueName());
                    if( queue == null){
                        System.out.println("[VirtualHost] 发送信息时发现队列不存在！ queueName = " + binding.getQueueName());
                        continue;
                    }
                    //构造新消息
                        Message message = Message.createMessageWithId(routingKey,basicProperties,body);
                    //判定这个消息是否可以转发给绑定对应的队列
                        //如果是 fanout,所有绑定的队列都要转发
                        //如果是 topic，还要判定一下， bindingKey 和 routingKey 是否匹配
                    if(!router.route(exchange.getType(),binding,message)){
                        continue;
                    }
                    //转发给队列
                    sendMessage(queue,message);
                }
            }
            return true;
        }catch(Exception e){
            System.out.println("[VirtualHost] 发送信息失败");
            e.printStackTrace();
            return false;
        }
    }

    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
        //此处的发送消息就是，把消息写入硬盘和磁盘中
        int deliverMode = message.getDeliverMode();
        //deliverMode为1：不持久化；为2：持久化
        if(deliverMode == 2){
            diskDataCenter.sendMessage(queue,message);
        }
        memoryDataCenter.sendMessage(queue,message);

        //此处还需要一个通知消费者可以消费的逻辑
        consumerManager.notifyConsume(queue.getName());
    }


    //订阅消息
    //添加一个队列订阅者，当队列收到消息之后，就要把消息推送到对应的订阅者
    //consumerTag:消费者的身份标识
    //autoAck:消息被消费完成之后，应答的方式，为 true 自动应答,为 false 手动应答
    //consmuer：是一个回调函数，此处类型设定范围函数式接口，这样后续调用 basicConsume 并且传实参的时候，就可以写作 lamda的样子
    public boolean basicConsumer(String consumerTag,String queueName,boolean autoAk,Consumer consumer){
        //构造一个 ConsumerEnv 对象，把这个对应的队列找到，再把 Consumer 对象添加到该队列中
        queueName = virtualHostName +queueName;

        try {
            consumerManager.addConsumer(consumerTag,queueName,autoAk,consumer);
            System.out.println("[VirtualHost] basicConsume 成功！ queueName = " + queueName);
            return true;
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("[VirtualHost] basicConsume 失败！ queueName = " + queueName);
            return false;
        }
    }

    public boolean basicAck(String queueName,String messageId){
        queueName = virtualHostName + queueName;
        try{
            //1.获取到消息和队列
            Message message = memoryDataCenter.getMessage(messageId);
            System.out.println("[VirtualHost] getMessage result: " + (message != null ? message : "null"));
            if(message == null){
                throw new MqException("[VirtualHost] 要确认的消息不存在！messageLd = " + messageId);
            }

            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            System.out.println("[VirtualHost] getQueue result: " + (queue != null ? queue.getName() : "null"));
            if(queue == null){
                throw new MqException("[VirtualHost] 要确认的消息不存在！queueName = " + queueName);
            }
            //2.删除硬盘上数据
            if(message.getDeliverMode() == 2){
                diskDataCenter.deleteMessage(queue,message);
            }
            //3.删除消息中心的数据
            memoryDataCenter.deleteMessage(messageId);
            memoryDataCenter.removeMessageWaitAck(queueName, messageId);
            System.out.println("[VirtualHost] basicAck 成功！queueName = " + queueName +
                    "messageId = " + messageId);

            return true;
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("[VirtualHost] basicAck 失败！消息确认失败！queueName = "+ queueName +
                    "messageId = " + messageId);
            return false;
        }

    }


}
